package websockets

import (
	"Godzilla/common/etcd"
	"Godzilla/common/responses"
	"Godzilla/conf/databases"
	"Godzilla/enum/code"
	"Godzilla/enum/fileds"
	"Godzilla/models"
	"Godzilla/repositories"
	"Godzilla/tcp/proto"
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"github.com/gorilla/websocket"
	"github.com/labstack/echo/v4"
	clientv3 "go.etcd.io/etcd/client/v3"
	"gopkg.in/fatih/set.v0"
	"io"
	"log"
	"math/rand"
	"net"
	"net/http"
	"strconv"
	"sync"
	"time"
)

// Node 连接节点保存
type Node struct {
	userId      int64
	Conn 	    *websocket.Conn
	DataQueue   chan []byte
	GroupSets   set.Interface
	tcpAddress  []string
}

var (
	// ClientMap 用户id跟连接资源关系
	ClientMap map[int64]*Node = make(map[int64]*Node, 0)
	// RWLocker 读写锁
	RWLocker sync.RWMutex

	//etcd 客户端
	clientEtcd *clientv3.Client

	// ClientTcpArr tcp客户端数组
	ClientTcpArr []string

)

func Doing(c echo.Context) error {
	var (
		token  		string
		isAllow 	bool
		userId		int64
		receiveMsg  []byte
	)

	//获取url里面的token，判断是否可以建立连接 连接url:ws://127.0.0.1:8080/ws?token=1111
	//连接后文本消息发送格式: {"id":1,"userId":1,"dstId":2,"cmd":10,"media":1,"content":"hello"}
	token = c.QueryParam("token")
	if token == "" {
		return c.JSON(http.StatusOK, responses.Response(code.ParameterError, "token不能为空", ""))
	}
	userManager, err:= repositories.NewUserManager()
	if err != nil {
		return c.JSON(http.StatusOK, responses.Response(code.SocketError, err.Error(), "socket数据库连接错误"))
	}


	userId, isAllow = userManager.CheckToken(token)

	//允许跨域 并返回是否可以连接
	conn,err :=(&websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return isAllow
		},
	}).Upgrade(c.Response(), c.Request(),nil)
	if err!=nil{
		log.Println(err.Error())
		return c.JSON(http.StatusOK, responses.Response(code.SocketError, err.Error(), "socket连接错误"))
	}

	//做用户id跟socket资源的映射
	node := &Node{
		userId:    userId,
		Conn:      conn,
		DataQueue: make(chan []byte, 50),
		GroupSets: set.New(set.ThreadSafe),
	}

	RWLocker.Lock()
	ClientMap[userId]=node
	RWLocker.Unlock()

	//将用户的群添加到node里面
	AddGroupId(userId)

	//etcd记录用户信息

	if clientEtcd, err = etcd.NewEtcdCli(); err != nil {
		log.Println("获取etcd客户端失败:",err.Error())
	}

	ctx, _ := context.WithTimeout(context.Background(), 5 * time.Second)
	var localAddress string
	if len(ClientTcpArr) != 0 {
		rand.Seed(time.Now().Unix())
		localAddress = ClientTcpArr[rand.Intn(len(ClientTcpArr))]
	}

	_, err = clientEtcd.Put(ctx,"/inline/"+ strconv.Itoa(int(userId)),  localAddress)
	if err != nil {
		log.Println("etcd Put /inline/userId 报错:", err.Error())
	}

	//发送数据
	go SendGoroutines(node)
	//接收数据
	go ReceiveGoroutines(node)

	msg := &models.Message{
		Id:       0,
		UserId:   0,
		Cmd:      fileds.CMD_HEART,
		DstId:    userId,
		Media:    fileds.MEDIA_TYPE_TEXT,
		Content:  "",
		Pic:      "",
		Url:      "",
		Des:      "连接成功",
		CreateAt: time.Now().Format("2006-01-02 15:04:05"),
	}
	receiveMsg, _ = json.MarshalIndent(msg, "", " ")
	SendMessage(userId,[]byte(receiveMsg))

	return nil
}



// SendGoroutines 发送协程
func SendGoroutines(node *Node) {
	for {
		select {
		case data:= <-node.DataQueue:
			err := node.Conn.WriteMessage(websocket.TextMessage,data)
			if err!=nil{
				//当发送信息时，接受信息的对象断开后会接到err报错：websocket: close sent
				//删除内存中的其中map
				RWLocker.Lock()
				delete(ClientMap, node.userId)
				RWLocker.Unlock()
				log.Println("SendGoroutines 发送信号")
				log.Println(err.Error())
				return
			}
		}
	}
}

// ReceiveGoroutines 接收协程
func ReceiveGoroutines(node *Node)  {
	for{
		_,data,err := node.Conn.ReadMessage()
		if err!=nil{
			//当客户端断开时可以此处会err报：websocket: close 1005 (no status)
			RWLocker.Lock()
			delete(ClientMap, node.userId)
			RWLocker.Unlock()
			log.Println("ReceiveGoroutines 接收信号")
			log.Println(err.Error())
			return
		}

		fmt.Printf("recv<=%s",data)
		//获取接收消息用户的id
		Dispatch(data)
	}
}

// SendMessage 发送消息
func SendMessage(userId int64, msg []byte)  {
	RWLocker.RLock()
	node, ok := ClientMap[userId]
	RWLocker.RUnlock()
	if ok {
		node.DataQueue <- msg
	}
}

// Dispatch  解密出接收的消息跟接收消息的用户
func Dispatch(data []byte) {
	// websocket是message单位不需要做粘包处理,tcp需要做粘包处理

	msg := models.Message{}
	err := json.Unmarshal(data, &msg)
	if err!=nil{
		log.Println(err.Error())
		return
	}

	switch msg.Cmd {
	case fileds.CMD_SINGLE_MSG:
		//分布式发送消息处理
		DistributedSingleSendMessage(msg, data)
		//SendMessage(msg.DstId, data)
	case fileds.CMD_ROOM_MSG:
		// 群聊
		DistributedRoomSendMessage(msg, data)
		//
		//for _,v:= range ClientMap{
		//	if v.GroupSets.Has(msg.DstId){
		//		v.DataQueue<-data
		//	}
		//}
	}
}

// DispatchLocal  接收到tcp服务端发送到本地的数据
func DispatchLocal(data []byte) {
	// websocket是message单位不需要做粘包处理,tcp需要做粘包处理

	msg := models.Message{}
	err := json.Unmarshal(data, &msg)
	if err!=nil{
		log.Println(err.Error())
		return
	}

	switch msg.Cmd {
	case fileds.CMD_SINGLE_MSG:
		SendMessage(msg.DstId, data)
	case fileds.CMD_ROOM_MSG:
		// 群聊
		for _,v:= range ClientMap{
			if v.GroupSets.Has(msg.DstId){
				v.DataQueue<-data
			}
		}
	}
}

// DistributedRoomSendMessage 群发消息
func DistributedRoomSendMessage(msg models.Message, data []byte)  {
	//本地群发
	for _,v:= range ClientMap{
		if v.GroupSets.Has(msg.DstId){
			v.DataQueue<-data
		}
	}
	//然后再发上群
	//在其他机器则轮训向其他tcp服务端发送tcp消息
	for _, v := range ClientTcpMap{
		if v.isConn {
			v.DataQueue <- data
			log.Println("发送成功:", v.Address)

		} else {
			log.Println("tcp服务端已关闭:", v.Address)
		}
		return
	}
}

// DistributedSingleSendMessage 单对单发消息
func DistributedSingleSendMessage(msg models.Message, data []byte)  {
	var (
		err  error
		getResponse *clientv3.GetResponse
	)

	//在本机直接发送
	RWLocker.RLock()
	node, ok := ClientMap[msg.DstId]
	RWLocker.RUnlock()
	if ok {
		node.DataQueue <- data
		return
	}
	//etcd 判断是否在线

	if clientEtcd, err = etcd.NewEtcdCli(); err != nil {
		log.Println("获取etcd客户端报错:", err.Error())
	}
	if err != nil {
	}
	ctx, _ := context.WithTimeout(context.Background(), 5 * time.Second)
	getResponse, err = clientEtcd.Get(ctx, "/inline/" + strconv.Itoa(int(msg.DstId)))
	if err != nil {
		log.Println("etcd获取用户所在客户端失败:", err.Error())
	}
	//不在线直接返回
	if len(getResponse.Kvs) == 0 {
		log.Println("etcd 未找到用户或不在线:", msg.DstId)
		return
	}
	//查找的目标地址 本机地址应该是从 CliTcpArr里面比较
	targetAddress := getResponse.Kvs[0].Value
	for _, v := range ClientTcpArr{
		if string(targetAddress) == v {
			log.Println("用户在本机，所以不发送到服务端;本机地址:", v, ";etcd存储的地址:", targetAddress)
			return
		}
	}
	//在其他机器则轮训向其他tcp服务端发送tcp消息
	for _, v := range ClientTcpMap{
		count := 0
		if count == 1 {
			return
		}
		if v.isConn {
			//向tcp协程发送未编码的数据

			v.DataQueue <- data
			log.Println("发送到tcp发送协程成功:", v.Address)
		} else {
			log.Println("tcp服务端已关闭:", v.Address)
		}
		return
	}

}

// AddGroupId 发送群消息前获取群信息添加在线群员
func AddGroupId(userId int64){
	//获取用户全部群Id
	contact, _ := repositories.NewContactManager()
	groupIds := contact.SearchGroupIds(userId)
	//将id放入连接存取里面
	//添加gid到set
	log.Println(groupIds)



	clientEtcd, err := etcd.NewEtcdCli()
	if err != nil {
		log.Println(err.Error())
	}
	ctx, _ := context.WithTimeout(context.Background(), 5 * time.Second)

	if len(groupIds) > 0 {
		for _, v := range groupIds {
			//取得node
			RWLocker.Lock()
			node,ok := ClientMap[userId]
			if ok{
				node.GroupSets.Add(v)
			}
			//clientMap[userId] = node
			//添加到etcd
			var localAddress string
			if len(ClientTcpArr) != 0 {
				rand.Seed(time.Now().Unix())
				localAddress = ClientTcpArr[rand.Intn(len(ClientTcpArr))]
			}
			_, err := clientEtcd.Put(ctx,"/group/group-"+ strconv.Itoa(int(v)) +"-"+strconv.Itoa(int(userId)), localAddress)
			if err != nil {
				log.Println("etcd Put /inline/userId 报错:", err.Error())
			}
			RWLocker.Unlock()
		}
	} else {
		RWLocker.Lock()
		node,ok := ClientMap[userId]
		if ok {
			//删除全部
			node.GroupSets = set.New(set.ThreadSafe)
		}
		RWLocker.Unlock()
	}
}

// GetClientMaps 查看当前服务器在线连接websocket的id
func GetClientMaps() []int64 {
	var (
		ids   []int64
	)
	for k, _ := range ClientMap {
		ids = append(ids, k)
	}
	return ids
}


//------------------------------------------------
//tcp客户端操作


//初始化
func init()  {

	go InitTcpClient()
}

type NodeTcp struct {
	Address    string
	Conn 	   *net.TCPConn
	DataQueue  chan []byte
	isConn     bool
}

var (
	ClientTcpMap map[string]*NodeTcp = make(map[string]*NodeTcp, 0)
)

func InitTcpClient()  {
	var (
		tcpServerArr 	[]string
	)

	//读取tcpServer的配置列表
	//读取tcpServer列表配置
	databases.TcpArrSetConfig()
	tcpServerArr = databases.GetTcpArrConfig()
	if len(tcpServerArr) == 0 {
		//没有配置tcp服务端直接退出
		return
	}


	//循环启动
	for _, address := range tcpServerArr {
		//服务器地址
		hawkServer, err := net.ResolveTCPAddr("tcp", address)
		if err != nil {
			log.Println("ResolveTCPAddr报错:", err.Error())
		}
		conn, err := net.DialTCP("tcp", nil, hawkServer)
		if err != nil {
			log.Println("连接服务器错误:", err.Error())
			return
		}
		ClientTcpArr = append(ClientTcpArr, conn.LocalAddr().String())
		log.Println(conn.LocalAddr().String())

		node := &NodeTcp{
			Address:   address,
			Conn:      conn,
			DataQueue: make(chan []byte, 128),
			isConn:    true,
		}

		RWLocker.Lock()
		ClientTcpMap[address] = node
		RWLocker.Unlock()
		
		go TcpClientSend(node)
		
		go TcpClientReceive(node)

	}
}

// TcpClientSend 发送到tcp服务端
func TcpClientSend(node *NodeTcp)  {
	timer := time.NewTimer(5 * time.Second)
	defer timer.Stop()
	for  {
		select {
		case data:= <-node.DataQueue:

			b, err := proto.Encode(string(data))
			if err != nil {
				fmt.Println("Encode失败，err:", err)
				return
			}
			node.Conn.Write(b)
			timer.Reset(10 * time.Second)

		case <-timer.C:
			log.Println("超时10秒发送的心跳")
			if node.isConn {

				msg := models.Message{
					Id:       0,
					UserId:   0,
					Cmd:      fileds.CMD_HEART,
					DstId:    0,
					Media:    0,
					Content:  "",
					Pic:      "",
					Url:      "",
					Des:      "",
					CreateAt: time.Now().Format("2006-01-02 15:04:05"),
				}
				msgHeart, _ := json.MarshalIndent(msg, "", " ")
				b, err := proto.Encode(string(msgHeart))
				if err != nil {
					fmt.Println("Encode失败，err:", err)
				}
				node.Conn.Write(b)
				timer.Reset(10 * time.Second)
			}
		default:
			//log.Println("执行default")
		}
	}
}

// TcpClientReceive 接收tcp服务端消息
func TcpClientReceive(node *NodeTcp)  {
	for  {
		if !node.isConn {
			log.Println("节点已关闭:", node.Address)
			return
		}

		reader := bufio.NewReader(node.Conn)
		msg, err := proto.Decode(reader)

		fmt.Println("收到服务端消息消息：", msg)
		if err == io.EOF {
			return
		}
		if err != nil {
			fmt.Println("decode失败，err:", err)
			return
		}
		if err == io.EOF {
			//TODO 同时把服务端ClientMap删除
			log.Println("tcp服务端关闭了")
			return
		}

		// 解码后发送到websocket客户端
		DispatchLocal([]byte(msg))

	}
}






